package com.feidee.fd.sml.algorithm.component.ml.classification

import com.feidee.fd.sml.algorithm.component.ml.MLParam
import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.classification.MultilayerPerceptronClassifier
import org.apache.spark.ml.linalg.Vectors

/**
  * @author YongChen, songhaicheng
  * @date 2018/3/21 10:18
  * @description songhaicheng
  * @reviewer songhaicheng
  */
case class MultilayerPerceptronParam (
                                       override val input_pt: String,
                                       override val output_pt: String,
                                       override val hive_table: String,
                                       override val flow_time: String,
                                       override var labelCol: String,
                                       override val featuresCol: String,
                                       override var predictionCol: String,
                                       override val modelPath: String,
                                       override val metrics: Array[String],
                                       // 指定神经网络的图层  比如 [8, 9, 8, 2],就是输入层 8 个节点(即 8 个特征)；两个隐藏层，隐藏结点数分别为 9 和 8；输出层 2 个结点(即二分类)，必填参数，长度大于 1，且每个值必须大于 0
                                       layers: Array[Int],
                                       // 用于在矩阵中堆叠输入数据的块大小以加速计算，> 0，默认128
                                       blockSize: Int,
                                       // 算法优化方法，支持 [gd, l-bfgs]，默认 l-bfgs
                                       solver: String,
                                       // 最大迭代次数，>= 0，默认 100
                                       maxIter: Int,
                                       // 优化算法迭代求解过程的收敛阀值，>= 0，默认 1E-6
                                       tol: Double,
                                       // 随机种子，默认 123456L
                                       seed: Long,
                                       // 初始化权重
                                       initialWeights: Array[Double],
                                       // 每次优化的迭代步长，仅适用于 solver = gd，> 0，默认 0.03
                                       stepSize: Double
                                     ) extends MLParam {

  def this() = this(null, null, null, null, "features", "label", "prediction", null, new Array[String](0),
    Array.empty[Int], 128, "l-bfgs", 100, 1E-6, 0L, Array.empty[Double], 0.03)

  override def verify(): Unit = {
    super.verify()
    require(layers.nonEmpty && layers.length > 1, "param layers can't be null and size > 1")
    require(blockSize > 0,"param blockSize can't be negative")
    require(maxIter >= 0, "param maxIter can't be negative")
    require(stepSize > 0, "param stepSize must be greater than 0")
    require(tol >= 0, "param tol can't be negative")
    val solvers = Array("gd", "l-bfgs")
    require(solvers.contains(solver.toLowerCase), s"param solver only accepts ${solvers.mkString("[", ", ", "]")}," +
      s" but has $solver")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("layers" -> layers)
    map += ("blockSize" -> blockSize)
    map += ("solver" -> solver)
    map += ("maxIter" -> maxIter)
    map += ("tol" -> tol)
    map += ("seed" -> seed)
    map += ("initialWeights" -> initialWeights)
    map += ("stepSize" -> stepSize)
    map
  }

}


class MultilayePerceptronComponent extends  AbstractClassificationComponent[MultilayerPerceptronParam] {
  /**
    * 根据参入参数设置算法组件实体
    *
    * @param param 运行参数
    * @return 算法组件
    */
  override def setUp(param: MultilayerPerceptronParam): PipelineStage = {
    val mpp=new MultilayerPerceptronClassifier()
      .setLabelCol(param.labelCol)
      .setFeaturesCol(param.featuresCol)
      .setPredictionCol(param.predictionCol)
      .setLayers(param.layers)
      .setBlockSize(param.blockSize)
      .setSolver(param.solver)
      .setMaxIter(param.maxIter)
      .setTol(param.tol)
      .setStepSize(param.stepSize)
      .setSeed(param.seed)

    if (param.initialWeights.nonEmpty) {
      mpp.setInitialWeights(Vectors.dense(param.initialWeights))
    }

     mpp
  }

}

object MultilayePerceptronComponent {

  def apply(paramStr: String): Unit = {
    new MultilayePerceptronComponent()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    MultilayePerceptronComponent(args(0))
  }
}



